package com.dianwoba.core.redis;

import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.core.TimeoutUtils;
import org.springframework.util.Assert;

import com.dianwoba.core.redis.operation.HashOperations;
import com.dianwoba.core.redis.operation.ListOperations;
import com.dianwoba.core.redis.operation.SetOperations;
import com.dianwoba.core.redis.operation.ValueOperations;
import com.dianwoba.core.redis.operation.ZSetOperations;
import com.dianwoba.core.redis.operation.impl.DefaultHashOperations;
import com.dianwoba.core.redis.operation.impl.DefaultListOperations;
import com.dianwoba.core.redis.operation.impl.DefaultSetOperations;
import com.dianwoba.core.redis.operation.impl.DefaultValueOperations;
import com.dianwoba.core.redis.operation.impl.DefaultZSetOperations;
import com.dianwoba.core.redis.serializer.JdkSerializationRedisSerializer;
import com.dianwoba.core.redis.serializer.RedisSerializer;
import com.dianwoba.core.redis.serializer.StringRedisSerializer;

import redis.clients.jedis.BinaryJedisCommands;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.util.Pool;

public class JedisTemplate<K, V, X extends BinaryJedisCommands> implements InitializingBean {

	private RedisSerializer<?> defaultSerializer = new JdkSerializationRedisSerializer();
	
	@SuppressWarnings("rawtypes")
	private RedisSerializer keySerializer = null;
	@SuppressWarnings("rawtypes")
	private RedisSerializer valueSerializer = null;
	@SuppressWarnings("rawtypes")
	private RedisSerializer hashKeySerializer = null;
	@SuppressWarnings("rawtypes")
	private RedisSerializer hashValueSerializer = null;
	private RedisSerializer<String> stringSerializer = new StringRedisSerializer();

	private ValueOperations<K, V> valueOps;
	private ListOperations<K, V> listOps;
	private SetOperations<K, V> setOps;
	private ZSetOperations<K, V> zSetOps;

	private Pool<X> jedisPool;

	/**
	 * Constructs a new <code>JedisTemplate</code> instance.
	 */
	public JedisTemplate() {
	}

	public JedisTemplate(Pool<X> jedisPool) {
		this.jedisPool = jedisPool;
	}

	protected X getResource() {
		return jedisPool.getResource();
	}

	protected void returnResource(X jedis) {
		if (jedis != null) {
			jedisPool.returnResource(jedis);
		}
	}
	
	protected void returnBrokenResource(X jedis) {
		if (jedis != null) {
			jedisPool.returnBrokenResource(jedis);
		}
	}

	public <T> T execute(JedisCallback<T, X> callback) {
		if (callback != null) {
			X jedis = null;
			try {
				jedis = getResource();
				return callback.doInRedis(jedis);
			} catch(JedisConnectionException ex) {
				returnBrokenResource(jedis);
				jedis = null;
			}
			finally {
				returnResource(jedis);
			}
		}
		return null;
	}

	public void afterPropertiesSet() {
		Assert.notNull(jedisPool, "jedisPool is required");
		if (keySerializer == null) {
			keySerializer = defaultSerializer;
		}
		if (valueSerializer == null) {
			valueSerializer = defaultSerializer;
		}
		if (hashKeySerializer == null) {
			hashKeySerializer = defaultSerializer;
		}
		if (hashValueSerializer == null) {
			hashValueSerializer = defaultSerializer;
		}
	}
	
	public ValueOperations<K, V> opsForValue() {
		if (valueOps == null) {
			valueOps = new DefaultValueOperations<K, V, X>(this);
		}
		return valueOps;
	}

	public ListOperations<K, V> opsForList() {
		if (listOps == null) {
			listOps = new DefaultListOperations<K, V, X>(this);
		}
		return listOps;
	}

	public SetOperations<K, V> opsForSet() {
		if (setOps == null) {
			setOps = new DefaultSetOperations<K, V, X>(this);
		}
		return setOps;
	}

	public ZSetOperations<K, V> opsForZSet() {
		if (zSetOps == null) {
			zSetOps = new DefaultZSetOperations<K, V, X>(this);
		}
		return zSetOps;
	}

	public <HK, HV> HashOperations<K, HK, HV> opsForHash() {
		return new DefaultHashOperations<K, HK, HV, X>(this);
	}

	public RedisSerializer<?> getDefaultSerializer() {
		return defaultSerializer;
	}

	public void setDefaultSerializer(RedisSerializer<?> defaultSerializer) {
		this.defaultSerializer = defaultSerializer;
	}

	public RedisSerializer<?> getKeySerializer() {
		return keySerializer;
	}

	public void setKeySerializer(RedisSerializer<?> keySerializer) {
		this.keySerializer = keySerializer;
	}

	public RedisSerializer<?> getValueSerializer() {
		return valueSerializer;
	}

	public void setValueSerializer(RedisSerializer<?> valueSerializer) {
		this.valueSerializer = valueSerializer;
	}

	public RedisSerializer<?> getHashKeySerializer() {
		return hashKeySerializer;
	}

	public void setHashKeySerializer(RedisSerializer<?> hashKeySerializer) {
		this.hashKeySerializer = hashKeySerializer;
	}

	public RedisSerializer<?> getHashValueSerializer() {
		return hashValueSerializer;
	}

	public void setHashValueSerializer(RedisSerializer<?> hashValueSerializer) {
		this.hashValueSerializer = hashValueSerializer;
	}

	public RedisSerializer<String> getStringSerializer() {
		return stringSerializer;
	}

	public void setStringSerializer(RedisSerializer<String> stringSerializer) {
		this.stringSerializer = stringSerializer;
	}

	public Pool<X> getJedisPool() {
		return jedisPool;
	}

	public void setJedisPool(Pool<X> jedisPool) {
		this.jedisPool = jedisPool;
	}
	
	@SuppressWarnings("unchecked")
	private byte[] rawKey(Object key) {
		Assert.notNull(key, "non null key required");
		if (keySerializer == null && key instanceof byte[]) {
			return (byte[]) key;
		}
		return keySerializer.serialize(key);
	}
	
	public void delete(K key) {
		final byte[] rawKey = rawKey(key);
		execute(new JedisCallback<Object, X>() {

			public Object doInRedis(X jedis) {
				jedis.del(rawKey);
				return null;
			}
		});
	}

	public Boolean expire(K key, final long timeout, final TimeUnit unit) {
		final byte[] rawKey = rawKey(key);
		final long rawTimeout = TimeoutUtils.toMillis(timeout, unit);

		return execute(new JedisCallback<Boolean, X>() {

			public Boolean doInRedis(X jedis) {
				try {
					return convert(jedis.pexpire(rawKey, rawTimeout));
				} catch (Exception e) {
					// Driver may not support pExpire or we may be running on Redis 2.4
					return convert(jedis.expire(rawKey, (int)TimeoutUtils.toSeconds(timeout, unit)));
				}
			}
		});
	}
	
	public Boolean convert(Long result) {
		return result != null ? result == 1 : null;
	}

}
